package com.smile.frame.mq.kafka.controller;

import cn.hutool.json.JSONUtil;
import com.smile.frame.mq.kafka.dto.OrderDTO;
import io.swagger.annotations.Api;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.requests.LeaveGroupRequest;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.lang.NonNull;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import javax.annotation.Resource;
import java.util.concurrent.ExecutionException;

/**
 * @author LGC
 */
@Slf4j
@Api(tags = "test")
@RestController
@RequestMapping("/kafka")
public class KafkaController {

    @Resource
    private KafkaTemplate<String, String> kafkaTemplate;

    @GetMapping("/send")
    public void send() throws ExecutionException, InterruptedException {
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId(1);
        orderDTO.setProductId(1);
        orderDTO.setNum(100);
        orderDTO.setPrice(100000);
        ProducerRecord<String, String> record = new ProducerRecord<>("test", String.valueOf(orderDTO.getOrderId()), JSONUtil.toJsonStr(orderDTO));
        // get为同步
        SendResult<String, String> result = kafkaTemplate.send(record).get();
        log.info("同步发送 result:{}", result);
    }


    @GetMapping("/asyncSend")
    public void asyncSend() {
        OrderDTO orderDTO = new OrderDTO();
        orderDTO.setOrderId(1);
        orderDTO.setProductId(1);
        orderDTO.setNum(100);
        orderDTO.setPrice(100000);
        ProducerRecord<String, String> record = new ProducerRecord<>("test", String.valueOf(orderDTO.getOrderId()), JSONUtil.toJsonStr(orderDTO));
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(record);
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable ex) {
                log.error("异步失败", ex);
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("异步成功:{}", result.getProducerRecord().key());
            }
        });
    }

    @GetMapping("/sendBatch")
    public void sendBatch() {
        for (int i = 0; i < 10; i++) {
            OrderDTO orderDTO = new OrderDTO();
            orderDTO.setOrderId(i);
            orderDTO.setProductId(i);
            orderDTO.setNum(100);
            orderDTO.setPrice(100000);
            kafkaTemplate.send("test1", String.valueOf(orderDTO.getOrderId()), JSONUtil.toJsonStr(orderDTO));
        }
    }

}